package com.ljx.mq.work;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 工作队列模式
 */
public class Consumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        // 创建Connection对象
        Connection connection = connectionFactory.newConnection();
        // 创建Channel
        Channel channel = connection.createChannel();
        // 设置队列属性
        /**
         * 第一个参数：队列名称
         * 第二个参数：队列是否持久化【重启后就消失了】
         * 第三个参数：是否排他性【只能在同一个连接对象中操作队列】
         * 第四个参数：是否自动删除【如果没有消费者连接就自动删除】
         * 第五个参数：是否需要设置其他参数
         */
        channel.queueDeclare("queue-study", false, false, false, null);
        // 预取，每次取1条消息
        channel.basicQos(1);
        // 从channel中取消息进行消费
        /**
         * 第一个参数：队列名称
         * 第二个参数：是否自动签收
         */
        // 关闭自动签收：为了取消负载均衡
        channel.basicConsume("queue-study", false, new DeliverCallback() {
            /**
             * 当消息取出就调用这个方法
             */
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("取出消息：" + new String(delivery.getBody()));
                // 手动签收
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }, new CancelCallback() {
            /**
             * 当消息取消了就调用这个方法
             */
            @Override
            public void handle(String s) throws IOException {
                System.out.println("取消消息");
            }
        });
    }
}
